package com.cx.config;

import com.cx.async.AsyncRedis;
import com.cx.exception.CacheConfigException;
import com.cx.property.LettuceProperties;
import com.cx.reactive.ReactiveRedis;
import com.cx.repository.SecurecyPostProcessor;
import com.cx.support.LettuceConnectionManager;
import com.cx.support.RedisLettuceCacheConfig;
import com.cx.support.SpringContextHolder;
import com.cx.utils.SerializeAdapter;
import com.zaxxer.hikari.HikariDataSource;
import io.lettuce.core.*;
import io.lettuce.core.api.StatefulConnection;
import io.lettuce.core.cluster.ClusterClientOptions;
import io.lettuce.core.cluster.ClusterTopologyRefreshOptions;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.masterslave.MasterSlave;
import io.lettuce.core.masterslave.StatefulRedisMasterSlaveConnection;
import io.lettuce.core.resource.ClientResources;
import io.lettuce.core.resource.DefaultClientResources;
import io.lettuce.core.support.AsyncConnectionPoolSupport;
import io.lettuce.core.support.AsyncPool;
import io.lettuce.core.support.BoundedPoolConfig;
import io.lettuce.core.support.ConnectionPoolSupport;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.jdbc.DataSourceBuilder;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.Primary;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.annotation.EnableTransactionManagement;

import javax.sql.DataSource;
import java.sql.SQLException;
import java.time.Duration;
import java.util.List;

/**
 * 文档参考：https://lettuce.io/core/5.1.7.RELEASE/reference/
 * @Author: 舒建辉
 * @Description:
 * @Date: Created on 2019/3/13
 */
@Configuration
@Import({SpringContextHolder.class,
        SecurecyPostProcessor.class,
        AsyncRedis.class,
        ReactiveRedis.class,
        LettuceRedisAdapter.class})
@EnableTransactionManagement
public class CacheTxAutoConfiguration {

    @Autowired
    private LettuceProperties lettuceProperties;

    @Primary
    @Bean
    public LettuceProperties lettuceProperties(){
        return new LettuceProperties();
    }

    //destroyMethod: 防止bean销毁时自动触发该方法
    @Bean(destroyMethod = "shutdown")
    public ClientResources clientResources() {

        return DefaultClientResources.builder()
                .ioThreadPoolSize(lettuceProperties.getIoThreadPoolSize())
                .computationThreadPoolSize(lettuceProperties.getComputationThreadPoolSize())
                .build();
    }

    @Bean
    public RedisLettuceCacheConfig redisLettuceCacheConfig() {

        String mode = null;
        ReadFrom readFrom = ReadFrom.MASTER_PREFERRED;
        if (lettuceProperties.getCluster() != null) {
            mode = lettuceProperties.getCluster().getMode();
            String readFromStr = lettuceProperties.getCluster().getReadFromStr();
            if (StringUtils.isNoneBlank(readFromStr)) {
                readFrom = ReadFrom.valueOf(readFromStr.trim());
            }
        }

        AbstractRedisClient client;
        StatefulConnection connection = null;
        AsyncPool asyncPool = null;

       List<RedisURI> uriList = lettuceProperties.clusterNodes();
        if (uriList.size() == 1) {
            RedisURI uri = uriList.get(0);
            if (readFrom == null) {
                client = RedisClient.create(clientResources(), uri);
                ((RedisClient) client).setOptions(buildClientOptions());
            } else {
                client = RedisClient.create(clientResources());
                ((RedisClient) client).setOptions(buildClientOptions());
                StatefulRedisMasterSlaveConnection c = MasterSlave.connect(
                        (RedisClient) client, LettuceRedisAdapter.objectGZIPCodec(), uri);
                c.setReadFrom(readFrom);
                connection = c;
            }
            asyncPool = asyncPool(client);
        } else {
            if (mode != null && mode.equalsIgnoreCase("MasterSlave")) {
                client = RedisClient.create(clientResources());
                ((RedisClient) client).setOptions(buildClientOptions());
                StatefulRedisMasterSlaveConnection c = MasterSlave.connect(
                        (RedisClient) client, LettuceRedisAdapter.objectGZIPCodec(), uriList);
                if (readFrom != null) {
                    c.setReadFrom(readFrom);
                }
                connection = c;
            } else {
                client = RedisClusterClient.create(clientResources(), uriList);
                ((RedisClusterClient) client).setOptions(buildClusterClientOptions());
                if (readFrom != null) {
                    StatefulRedisClusterConnection c = ((RedisClusterClient) client).connect(LettuceRedisAdapter.objectGZIPCodec());
                    c.setReadFrom(readFrom);
                    connection = c;
                }
            }
            asyncPool = asyncPool(client);
        }

        LettuceConnectionManager m = LettuceConnectionManager.defaultManager();
        m.init(client, connection, asyncPool);

        RedisLettuceCacheConfig redisLettuceCacheConfig = new RedisLettuceCacheConfig();
        redisLettuceCacheConfig.setManager(m);
        redisLettuceCacheConfig.setAsyncPool(asyncPool);
        redisLettuceCacheConfig.setConnection(connection);
        redisLettuceCacheConfig.setRedisClient(client);
        redisLettuceCacheConfig.setAsyncResultTimeoutInMillis(lettuceProperties.getAsyncResultTimeoutInMillis());

        return redisLettuceCacheConfig;
    }

    // 异步连接池
    private AsyncPool asyncPool(AbstractRedisClient redisClient) {
        if (redisClient instanceof RedisClient) {
            if (lettuceProperties.isOpenPipeline()) {
                /**
                 * 异步发布订阅连接池
                 */
                return AsyncConnectionPoolSupport.createBoundedObjectPool(
                        () -> ((RedisClient) redisClient).connectPubSubAsync(LettuceRedisAdapter.objectGZIPCodec(), lettuceProperties.getRedisURI()),
                        SerializeAdapter.boundedPoolConfig());
            } else {
                /**
                 * 异步连接池
                 */
                return AsyncConnectionPoolSupport.createBoundedObjectPool(
                        () -> ((RedisClient) redisClient).connectAsync(LettuceRedisAdapter.objectGZIPCodec(), lettuceProperties.getRedisURI()),
                        SerializeAdapter.boundedPoolConfig());
            }
        } else if (redisClient instanceof RedisClusterClient) {

            if (lettuceProperties.isOpenPipeline()) {
                /**
                 * 异步集群发布订阅连接池
                 */
                return AsyncConnectionPoolSupport.createBoundedObjectPool(
                        () -> ((RedisClusterClient) redisClient).connectPubSubAsync(LettuceRedisAdapter.objectGZIPCodec()),
                        SerializeAdapter.boundedPoolConfig());
            } else {
                /**
                 * 异步集群专用连接池
                 */
                return AsyncConnectionPoolSupport.createBoundedObjectPool(
                        () -> ((RedisClusterClient) redisClient).connectAsync(LettuceRedisAdapter.objectGZIPCodec()),
                        SerializeAdapter.boundedPoolConfig());
            }
        } else {
            throw new CacheConfigException("type " + redisClient.getClass() + " is not supported");
        }
    }

    // 同步连接池
    private GenericObjectPool pool(AbstractRedisClient redisClient) {
        if (redisClient instanceof RedisClient) {
            if (lettuceProperties.isOpenPipeline()) {
                /**
                 * 同步发布订阅连接池
                 */
                return ConnectionPoolSupport.createGenericObjectPool(() ->
                        ((RedisClient) redisClient).connectPubSub(LettuceRedisAdapter.objectGZIPCodec(), lettuceProperties.getRedisURI()), SerializeAdapter.poolConfig());
            } else {
                /**
                 * 同步连接池
                 */
                return ConnectionPoolSupport.createGenericObjectPool(() ->
                        ((RedisClient) redisClient).connect(LettuceRedisAdapter.objectGZIPCodec(), lettuceProperties.getRedisURI()), SerializeAdapter.poolConfig());
            }
        } else if (redisClient instanceof RedisClusterClient) {
            if (lettuceProperties.isOpenPipeline()) {
                /**
                 * 同步集群发布订阅连接池
                 */
                return ConnectionPoolSupport.createGenericObjectPool(() ->
                        ((RedisClusterClient) redisClient).connectPubSub(LettuceRedisAdapter.objectGZIPCodec()), SerializeAdapter.poolConfig());
            } else {
                /**
                 * 同步集群专用连接池
                 */
                return ConnectionPoolSupport.createGenericObjectPool(() ->
                        ((RedisClusterClient) redisClient).connect(LettuceRedisAdapter.objectGZIPCodec()), SerializeAdapter.poolConfig());
            }
        } else {
            throw new CacheConfigException("type " + redisClient.getClass() + " is not supported");
        }
    }

    /**
     * 客户端选择器
     * @return
     */
    private ClientOptions buildClientOptions() {
        return ClientOptions.builder()
                .pingBeforeActivateConnection(false)
                .autoReconnect(true)
                .cancelCommandsOnReconnectFailure(false)
                .suspendReconnectOnProtocolFailure(false)
                .requestQueueSize(lettuceProperties().getRequestQueueSize())
                .disconnectedBehavior(ClientOptions.DisconnectedBehavior.REJECT_COMMANDS)
                .timeoutOptions(TimeoutOptions.enabled(Duration.ofMinutes(lettuceProperties().getTimeoutOptions())))  //全局命令操作超时时间
                .build();
    }

    /**
     * 集群客户端选择器
     * @return
     */
    private ClusterClientOptions buildClusterClientOptions() {
        return ClusterClientOptions.builder()
                .topologyRefreshOptions(ClusterTopologyRefreshOptions.builder()
                        .enablePeriodicRefresh(Duration.ofSeconds(lettuceProperties().getCacheLevel1Ttl()))
                        .enableAllAdaptiveRefreshTriggers()
                        .build())
                .build();
    }

    /**
     * 通用连接池配置
     * test-on-borrow: false #当调用borrow Object方法时，是否进行有效性检查(生产环境一般不开启，影响性能)
     * test-on-return: false #当调用return Object方法时，是否进行有效性检查(生产环境一般不开启，影响性能)
     * test-while-idle: true #Idle时进行连接扫描
     * time-between-eviction-runs-millis: 30000 #表示idle object evitor两次扫描之间要sleep的毫秒数
     * num-tests-per-eviction-run: 10 #表示idle object evitor每次扫描的最多的对象数
     * min-evictable-idle-time-millis: 60000 #表示一个对象至少停留在idle状态的最短时间，然后才能被idle object evitor扫描并驱逐；这一项只有在timeBetweenEvictionRunsMillis大于0时才有意义
     * @return
     */
    @Bean
    public GenericObjectPoolConfig poolConfig(){

        GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
        poolConfig.setMaxTotal(lettuceProperties.getPool().getMaxActive());
        poolConfig.setMaxIdle(lettuceProperties.getPool().getMaxIdle());
        poolConfig.setMinIdle(lettuceProperties.getPool().getMinIdle());
        poolConfig.setMaxWaitMillis(lettuceProperties.getPool().getMaxWait());
        poolConfig.setTestOnBorrow(false);
        poolConfig.setTestOnReturn(false);
        poolConfig.setTestWhileIdle(true);
        poolConfig.setTimeBetweenEvictionRunsMillis(30000);
        poolConfig.setNumTestsPerEvictionRun(10);
        poolConfig.setMinEvictableIdleTimeMillis(60000);

        return poolConfig;
    }

    /**
     * Lettuce连接池配置
     * @return
     */
    @Bean
    protected BoundedPoolConfig boundedPoolConfig(){

        return BoundedPoolConfig.builder()
                .maxTotal(lettuceProperties.getPool().getMaxActive())
                .maxIdle(lettuceProperties.getPool().getMaxIdle())
                .minIdle(lettuceProperties.getPool().getMinIdle())
                .testOnCreate(lettuceProperties.getPool().isTestOnCreate())
                .testOnAcquire(lettuceProperties.getPool().isTestOnAcquire())
                .testOnRelease(lettuceProperties.getPool().isTestOnRelease())
                .build();
    }

    @Bean(name="hikariDataSource")
    @Qualifier("hikariDataSource")
    @ConfigurationProperties(prefix="spring.datasource.hikari")
    public DataSource hikariDataSource() {

        return DataSourceBuilder.create().type(HikariDataSource.class).build();
    }

    /**
     * 缓存整合JPA事务
     * @return
     * @throws SQLException
     */
    @Bean
    public PlatformTransactionManager transactionManager() throws SQLException{

        return new DataSourceTransactionManager(hikariDataSource());
    }
}